package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

/**
 * 通过这个类，针对硬盘上的消息进行管理
 */
public class MessageFileManager {

    // 定义一个内部类，表示该队列的统计消息
    // 优先考虑使用 static,静态内部类
    static public class Stat {
        // 此处直接定义成 public
        public int totalCount;  // 总的消息数
        public int validCount; // 有效消息数

    }

    public void init(){
        // 此处暂时不需要做啥，方便以后扩展
    }

    // 约定消息文件所在的目录名字和文件名
    // 这个方法用来获取 指定队列对应的目录名字
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 这个方法用来获取该队列的消息数据文件路径
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "./queue_data.txt";
    }

    // 这个方法用来获取队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "./queue_stat.txt";
    }

    // 读取文件统计信息
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    // 写入统计信息到硬盘
    private void writeStat(String queueName, Stat stat) {
        // 使用 PrintWrite 来写文件
        // OutputStream 打开文件，默认情况下，会直接把源文件清空，此时就相当于 新数据把旧的数据覆盖了
        // 加个 参数 true，就会变成追加 new FileOutputStream(getQueueStatPath(queueName),true)
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建队列对应的文件目录
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 不存在就创建这个目录
            Boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败！baseDir=" + baseDir.getAbsolutePath());
            }
        }

        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            Boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败！ queueDateFile=" + queueDataFile.getAbsolutePath());
            }
        }

        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            Boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建统计文件失败！ queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }

        // 4. 给消息统计文件，设置初始值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    // 删除队列的目录和文件
    // 队列也是可以被删除的，当队列删除后，对应的目录文件，也需要随之删除
    public void destroyQueueFiles(String queueName) throws IOException {
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok1 = queueStatFile.delete();
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok2 = queueDataFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            throw new IOException("删除目录和文件失败！baseDir=" + baseDir.getAbsolutePath());
        }
    }

    // 检查队列的目录和文件是否存在
    // 比如生产者给 brokerserver 生产消息，这个消息就可能需要 记录到文件中（取决于是否持久化）
    public boolean checkFilesExists(String queueName) {
        // 判定队列的数据文件和统计文件是否都存在
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        return true;
    }

    // 这个方法是把新的消息，写入到对应文件中
    // queue 表示要把消息写入的队列是哪个， message表示要写入的消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1. 检查下当前要写入的队列 对应的文件是否存在
        if (!checkFilesExists(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在！queueName=" + queue.getName());
        }
        // 2. 把对象进行序列化，转换成二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);

        synchronized (queue) {
            // 3. 先获取到当前队列数据文件的长度，用这个长度来计算该 Message 对象和 offsetBeg offsetEnd
            // 把新的 message 数据，写入到队列的文件末尾，
            // 此时，message 对象的 offsetBeg 就是 当前文件长度+4
            // offsetEnd 就是 当前文件长度 + 4 + message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // 通过这个方法 queueDataFile.length() 就能获取到长度，单位字节
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 4. 写入消息数据到文件，注意，此处是追加
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 接下来首先写入的是当前消息的长度，占领4个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息本体
                    dataOutputStream.write(messageBinary);


                    // TODO
                }
            }

            // 5. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            // 重新写入
            writeStat(queue.getName(), stat);
        }

    }

    // 这是消息删除方法
    // 这里删除是逻辑删除，也就是把硬盘上的message对象里面的 isValid，设置成0
    // 1. 先把文件从硬盘中读取出来
    // 2. 然后修改 isValid
    // 3. 再写回到硬盘中
    // 此处这个参数中的 message对象，必须包含有效的 offsetBeg 和 offsetEnd
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 先读取对应 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 2. 读取当前的二进制数据，转换成 Message 对象 并修改 isValid
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 此处不需要给参数的 message 的 isValid 设置成0，因为这个参数是在内存中管理的 message对象，
                // 而这个对象 也要被马上删除了
                diskMessage.setIsValid((byte) 0x0);
                // 3. 写回去
                // 需要重新定位光标
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }

            // 还要更换统计文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }


    // 使用这个方法，从文件中，读取所有的消息内容，加载到内存中（具体来说是一个链表中）
    // 这个方法，准备在程序启动的时候，进行调用
    // 这里 使用一个 LinkedList，主要目的是为了后续进行头删操作
    // 这个方法的参数，只是一个 queueName，而不是 MsgQueue对象，因为不需要使用加锁
    // 不涉及多线程操作
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();

        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                // 手动记录光标的位置
                long currentOffset = 0;
                // 一个文件可能包含多个消息，所以要循环读
                while (true) {
                    // 1. 读取当前消息长度  , 一次读4个字节  (这里的 readInt 可能会读到文件的末尾)
                    // 读到末尾就会抛出 EOFException 异常
                    int messageSize = dataInputStream.readInt();
                    // 2. 按照这个长度，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        throw new MqException("[MessageFileManager] 文件格式错误！queueName=" + queueName);
                    }
                    // 3. 把读到的二进制数据，反序列化为 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 4. 判定这个消息，是不是无效对象 isValid = 0x0
                    if (message.getIsValid() != 0x1) {
                        // 无效数据，跳过
                        continue;
                    }
                    // 5. 有效数据，则需要把这个 Message 对象加入到链表中，加入之前要先设置 offsetBeg, offsetEnd
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 这个并非真是处理异常，处理正常业务逻辑
                // 文件读到末尾
                System.out.println("[MessageFileManager] 从硬盘恢复数据到内存完成！");
            }
        }
        return messages;
    }

    // 检查当前是否要针对该队列消息数据文件进行 GC
    public boolean checkGC(String queueName) {
        // 判定是否要GC,是根据总消息数和有效消息数，这两个值都在 消息统计文件中
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }

    // 约定新文件所在的位置
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    public void GC(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 此处要加锁，因为在进行GC时，不允许其他线程进行操作
        synchronized (queue) {
            // 由于 GC 操作比较耗时，我们统计一下消耗时间
            long gcBeg = System.currentTimeMillis();
            // 1. 创建一个新文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 正常情况下应该是不存在,如果存在，说明上一次GC一半，程序意外崩溃
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在！queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败！queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }
            // 2. 从旧的文件中，读取所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 3. 把有效数据写到新文件中
            try(OutputStream outputStream = new FileOutputStream(getQueueDataNewPath(queue.getName()))){
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for (Message message : messages){
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            // 4. 删除旧的数据文件，新文件重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧文件失败！queueOldDataFile="+ queueDataOldFile.getAbsolutePath());
            }
            // 把 queue_data_new.txt ==> queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败！queueDataNewFile="+ queueDataNewFile.getAbsolutePath());
            }

            // 5. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕！queueName=" + queue.getName()
                    + ", time=" + (gcEnd - gcBeg) + "ms");
        }
    }


}
